package day04

import java.net.URL

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object SubjectModulleCount {

  def main(args: Array[String]): Unit = {
    /**
      * 需求：求各个学科各个模块的访问量的top3
      * 实现思路：
      *   1.计算出每个学科的各个模块（url）的访问数量
      *   2.按照学科进行分组
      *   3.组内排序并取的top3
      */

    val conf = new SparkConf().setAppName("SubjectModuleCount").setMaster("local[2]")
    val sc: SparkContext = new SparkContext(conf)
    val logs: RDD[String] = sc.textFile("D://视频资料/大数据/spark/Day04/access.txt")

    //数据：http://java.learn.com/java/javaee.shtml   注意学科是http://java.learn.com，后面的是模块
    //将用户访问日志进行切片并返回url
    val url: RDD[String] = logs.map(line => line.split("\t")(1))
    //将url生成元祖，便于聚合
    val tupUrl: RDD[(String, Int)] = url.map((_,1))
    //获取每个学科的各个模块的访问量
    val reducedUrl: RDD[(String, Int)] = tupUrl.reduceByKey(_+_)
    //通过上面的数据来整合重新整合数据为二元组方便聚合：(学科subject,(url,count))
    val subjectAndUrlCount: RDD[(String, (String, Int))] = reducedUrl.map(x => {
      val url: String = x._1
      val count: Int = x._2
      //注意http://java.learn.com/java/javaee.shtml要想提出java.learn.com这个数据可以用java的URL的getHost方法
      //注意java提供了可以从一个url中获取Host的方法如下
      val subject: String = new URL(url).getHost
      (subject, (url, count))
    })
    //按照subject进行分组

    val grouped: RDD[(String, Iterable[(String, Int)])] = subjectAndUrlCount.groupByKey()
    //组内进行排序，注意Iterable类型必须转换成toList才能排序sortBy
    val sorted: RDD[(String, List[(String, Int)])] = grouped.mapValues(_.toList.sortBy(_._2).reverse)
    //获得top3
    val res: mutable.Buffer[(String, List[(String, Int)])] = sorted.mapValues(_.take(3)).collect().toBuffer
    res.foreach(x=>println("res = "+x))

  }

}
